// Copyright (C) 2010 and onwards Noam Shazeer and Georges Harik

#ifndef _TABLE_H_
#define _TABLE_H_

#include <sstream>
#include "util.h"
#include "file.h"
#include "thing.h"

// We want single-threaded resizable overlayable Hash Tables

// HashTable is a base class that resizes and is usable as a
// base for an overlayable class. It is not templated and stores
// uint64s

// Table is the overlayable class that is templated.
// Different data structures can be implemented on top of Table
// such as sets if integers and sets of things

// cell is empty
const uint64 kEmptyKey = Fingerprint("kEmptyKey");

// cell has been deleted.  Can be reused
const uint64 kDeletedKey = Fingerprint("kDeletedKey");

// Found in an overlay Table to indicate that a key present in the 
// base has been deleted in the overlay.
const uint64 kOverlayDeletedValue = Fingerprint("kOverlayDeletedValue");

struct HashTable {
  struct FullException {};
  struct Cell {
    uint64 key_;
    uint64 value_;
    string ToString() const { 
      return itoa(key_) + " " + itoa(value_);    }
    bool IsValid() const { return (key_ != kEmptyKey && key_ != kDeletedKey) ; }
  };
  Cell *alloc_;
  Cell *data_;
  uint64 capacity_; // a power of 2
  bool fallable_; // if not fallable, uses new to reallocate data.
  
  uint64 size_; // always a power of 2
  uint64 mask_; // size_ - 1

  // counts - cells are either empty, or contain values or are deleted. 
  int64 num_empty_;
  int64 num_deleted_;
  
  // this bool is set to true on reorg.  
  // Set it externally to false, do some stuff, and check it to see
  // if your references are still valid. 
  bool reorged_;         

  void Allocate(uint64 capacity) {
    capacity_ = capacity;
    alloc_ = new Cell[capacity_ + 1];
    data_ = alloc_ + 1;
  }

  HashTable(uint64 capacity, bool fallable) {
    CHECK(capacity > 0);
    CHECK(!(capacity & (capacity-1)));
    fallable_ = fallable;
    Allocate(capacity);
    Clear();
  }

  ~HashTable() {
    delete[] alloc_;
  }

  void Clear() {
    // cout << "::Clearing " << (void *)this << endl;
    SetSize(1);
    num_empty_ = 1;
    num_deleted_ = 0;
    data_[0].key_ = kEmptyKey; 
  }

  void SetSize(uint64 size) { 
    size_ = size; 
    mask_ = size_ - 1; 
    CHECK(!(size_ & mask_));
  }

  void Load(string fn){   
    Memblock block = File::Read(fn);
    alloc_ = (Cell*) block.data_;
    data_ = alloc_ + 1;
    SetSize(block.size() / sizeof(Cell) - 1);
    fallable_ = false; 
    capacity_ = size_;
    num_empty_ = *((uint64*) alloc_);
    num_deleted_ = *(((uint64*) alloc_) + 1);
  }
  
  void Store(string fn) {
    CHECK(!fallable_);
    uint64 * extra = (uint64*) alloc_;
    extra[0] = num_empty_; extra[1] = num_deleted_;
    File::Write(fn,
		Memblock((const char *)alloc_, (1 + size_) * sizeof(Cell)));
  }
  
  HashTable(string fn) { 
    Load(fn);
  }
  
  string ToString() {
    ostringstream o;
    o << "size_: " << size_ << endl
      << "capacity: " << capacity_ << endl
      << "fallable_: " << (fallable_ ? "T" : "F") << endl;
    for (uint64 i=0; i<size_; i++)
      o << data_[i].ToString() << endl;
    return o.str();
  }

  // returns the cell where key exists or would be inserted.
  Cell * Resolve(uint64 key) {
    uint64 pos = Fingerprint(key) & mask_;
    uint64 step = 0;
    Cell * first_deleted = NULL;
    while (1) {
      uint64 k = data_[pos].key_;
      if (k == key) return data_ + pos;
      if (k == kEmptyKey) 
	return (first_deleted ? first_deleted : (data_ + pos));
      if ( (k == kDeletedKey) && !first_deleted) first_deleted = data_ + pos;
      step++;
      pos = (pos + step) & mask_;
      if (step >  size_ * 2) {
	cout << "step: " << step
	     << " size_: " << size_
	     << " numValues: " << NumValues()
	     << " num_empty_: " << num_empty_
	     << " num_deleted_: " << num_deleted_
	     << " whole shebang\n" << ToString() << endl;
	CHECK(false);
      }
    }
  }
  
  const uint64 * Lookup(uint64 key) {
    Cell * look = Resolve(key);
    if (look->key_ == key) return &look->value_;
    return NULL;
  }
  
  // Inserts the key if absent with a 0 value
  uint64 * LookupWritable(uint64 key, bool *inserted) {
    *inserted = false;
    if (TooFull()) Reorg();
    Cell * look = Resolve(key);
    if (look->key_ == key) return &look->value_;
    if (look->key_ == kDeletedKey) num_deleted_--;
    else {
      CHECK(look->key_ == kEmptyKey);
      num_empty_--;
    }
    look->key_ = key;
    look->value_ = 0;
    *inserted = true;
    return &(look->value_);
  }

  void Delete(uint64 key) {
    Cell * look = Resolve(key);
    if (look->key_ == key) {
      look->key_ = kDeletedKey;
      num_deleted_++;
    }
  }

  bool TooFull() { return num_empty_ < size_ * 0.3 + 1; }

  uint64 NumValues() { return size_ - num_empty_ - num_deleted_;}
  uint64 CountNumValues() {
    uint64 ret = 0;
    for (uint64 i=0; i<size_; i++) if (data_[i].IsValid()) ret++;
    return ret;
  }
  void CheckNumValues() { CHECK(NumValues() == CountNumValues()); }
  uint64 CheckSum() {
    uint64 ret = 0;
    for (uint64 i=0; i<size_; i++) 
      if (data_[i].IsValid()) ret += data_[i].key_ * data_[i].value_;
    return ret;
  }
  void CheckDups() {
    set<uint64> seen;
    for (uint64 i=0; i<size_; i++) {
      if (data_[i].IsValid()) {	
	CHECK(!(seen % data_[i].key_));
	seen.insert(data_[i].key_);
	CHECK(Resolve(data_[i].key_) == data_ + i);
      }
    }
  }
  
  void InsertInternal(const Cell & c) {
    if ( (c.key_ != kEmptyKey) && (c.key_ != kDeletedKey)) {
      Cell * look = Resolve(c.key_);
      look->key_ = c.key_; look->value_ = c.value_;
    }    
  }

  void Reorg() {
    // cout << "Call to Reorg " << endl;
    reorged_ = true;
    CHECK(TooFull());
    uint64 num_values = NumValues();
    uint64 new_size = 1ull << CeilingLg2(num_values * 2 + 2);
    if (new_size > capacity_) {
      // need to allocate new space, or fail. 
      if (fallable_) throw FullException();
      Cell * old = data_;
      Cell * old_alloc = alloc_;
      //cout << "Allocating new space of size " << new_size << endl;
      Allocate(new_size);
      uint64 old_size = size_;
      for (uint64 i=0; i<new_size; i++) data_[i].key_ = kEmptyKey;
      SetSize(new_size);
      for (uint64 i=0; i<old_size; i++) InsertInternal(old[i]);
      delete [] old_alloc;
    } else {
      // Reorg in place
      for (uint64 i=0; i<size_; i++) 
	if (data_[i].key_ == kDeletedKey) data_[i].key_ = kEmptyKey;
      for (uint64 i=size_; i<new_size; i++) data_[i].key_ = kEmptyKey;
      uint64 old_size = size_;
      SetSize(new_size);
      // cout << "Reorg in place " << old_size << " " << new_size << endl;
      
      uint64 no_op_count = 0;
      uint64 i=0;
      uint64 count = 0;
      while (1) {
	count++;
	if (data_[i].IsValid()) {
	  Cell * target = Resolve(data_[i].key_);
	  if (target != data_ + i) {
	    target->key_ = data_[i].key_;
	    target->value_ = data_[i].value_;
	    data_[i].key_ = kEmptyKey;
	    no_op_count = 0;
	  }
	}
	i = (i+1) & mask_;
	no_op_count++;
	if (no_op_count > size_) break;
      }
      // cout << "Finished Reorg - count = " << count << endl;
      for (uint64 i=size_; i<old_size; i++) InsertInternal(data_[i]);
    }
    num_empty_ = size_ - num_values;
    num_deleted_ = 0;
  }
};

// overlayable hash table
template <class Value>
struct Table {
  HashTable h_;
  Table<Value> *base_;  
  
  Table(uint64 size, bool fallable, Table *base)
  : h_(size, fallable), base_(base) {}
  
  Table(string fn) :h_(fn), base_(NULL) {} // loads from file

  HashTable * GetHashTable() {
    CHECK(!base_);
    return &h_;
  }

  bool OverlayIsEmpty() { return (h_.NumValues() == 0); }

  void AddBase(Table<Value> *base) {
    base_ = base;
    h_.Clear();
  }

  void Store(string fn) { CHECK(!base_); h_.Store(fn); }
  
  const Value * Lookup(uint64 key) {
    return (const Value *)LookupI(key); }

  Value * LookupWritable(uint64 key, bool *inserted) {
    return (Value *)LookupWritableI(key, inserted); }

  // writes changes to base.  
  void Commit() {
    CHECK(base_);
    for (uint64 i=0; i<h_.size_; i++) {
      const HashTable::Cell & c = h_.data_[i];
      if (!c.IsValid()) continue;
      if (c.value_ == kOverlayDeletedValue) base_->Delete(c.key_);
      else (*base_)[c.key_] = c.value_;
    }
    h_.Clear();
  }

  const uint64 * LookupI(uint64 key) { 
    const uint64 * look = h_.Lookup(key);
    if (!base_) return look;
    if (look) {
      if (*look == kOverlayDeletedValue) return NULL;
      return look;
    }
    return base_->LookupI(key);
  }
  
  uint64 * LookupWritableI(uint64 key, bool *inserted) { 
    uint64 * look = h_.LookupWritable(key, inserted);
    if (!base_) return look;
    // there is a base and we are inserting if necessary
    if (*inserted) {
      // it wasn't there. look see if it was in the base. 
      const uint64 * look_base = base_->LookupI(key);
      if (look_base) {
	*look = *look_base;
	*inserted = false;
      }
      return look;
    } else {
      if ((uint64)(*look) == kOverlayDeletedValue) {
	*look = 0;
	*inserted = true;
      }
      return look;
    }
  }
  
  Value & operator [] (uint64 k) {
    bool inserted;
    return *LookupWritable(k, &inserted); }
  
  const Value * operator % (uint64 k) {
    return Lookup(k); }

  Value GetValue(uint64 k) {
    const Value * look = Lookup(k);
    if (look) return *look; return 0;
  }

  Value CheckGetValue(uint64 k) {
    const Value * look = Lookup(k);
    CHECK(look);
    return *look;
  }
  
  void Delete(uint64 k) {
    if (base_) {
      (*this)[k] = (Value)kOverlayDeletedValue;
      return;
    }
    h_.Delete(k);
  }
  
  string ToString() {
    ostringstream o;
    o << "OverlayHashTable " << endl;
    o << h_.ToString();
    if (base_) {
      o << "BASE:" << endl;
      o << base_->ToString();
    }
    return o.str();
  }
  
  void Clear() { 
    CHECK(!base_);
    h_.Clear(); 
  }
};

#endif
